package org.example.data_stream;

import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.connector.file.sink.FileSink;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * @author shenguangyang
 */
public class E11_SinkToFile {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        List<Event> events = new ArrayList<Event>() {{
            add(new Event("user02", "./cart", 2000L));
            add(new Event("user02", "./prod?id=10`", 3800L));
            add(new Event("user02", "./prod?id=2`", 3000L));
            add(new Event("user01", "./home1", 1000L));
            add(new Event("user01", "./prod?id=15`", 4000L));
            add(new Event("user03", "./prod?id=20`", 4000L));
            add(new Event("user04", "./prod?id=22`", 4000L));
            add(new Event("user05", "./prod?id=24`", 4000L));
        }};
        DataStream<Event> streamSource = env.fromCollection(events);

        final FileSink<String> sink = FileSink
                .forRowFormat(new Path("./output/"), new SimpleStringEncoder<String>("UTF-8"))
                .withRollingPolicy(
                        DefaultRollingPolicy.builder()
                                // 10s就会生产 一个新文件
                                .withRolloverInterval(Duration.ofSeconds(10))
                                // 10分钟没有数据来就开启一个新文件
                                .withInactivityInterval(Duration.ofSeconds(10))
                                // 文件最大大小
                                .withMaxPartSize(MemorySize.ofMebiBytes(1024 * 1024))
                                .build())
                .build();
        streamSource.map(Event::toString).sinkTo(sink).setParallelism(4);

        env.execute();
    }
}
